package com.niit.kafka;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

/**
 * 自定义日志处理器
 */
public class LogProcessor implements Processor<byte[],byte[]> {
    // Initialize this processor with the given context.
    private ProcessorContext processorContext;
    @Override
    public void init(ProcessorContext context) {
        this.processorContext = context;
    }
    // Process the record with the given key and value.
    @Override
    public void process(byte[] key, byte[] value) {
        System.out.println("value: " + new String(value));
        // 核心处理流程
        String input = new String(value);
        // 提取数据，以固定前缀过滤日志信息
        if( input.contains("PRODUCT_RATING_PREFIX:") ){
            System.out.println("product rating data coming! " + input);
            input = input.split("PRODUCT_RATING_PREFIX:")[1].trim();
            processorContext.forward("logProcessor".getBytes(), input.getBytes());
        }
    }
    @Override
    public void punctuate(long timestamp) {}
    @Override
    public void close() { }
}
